package com.sf.iisp.redis.listener;

import com.sf.iisp.BindingException;
import com.sf.iisp.bind.biz.BindingBiz;
import com.sf.iisp.bind.biz.UnbindingBiz;
import com.sf.iisp.bind.dao.IBindingDao;
import com.sf.iisp.bind.dao.IUnbindingDao;
import com.sf.iisp.bind.domain.*;
import com.sf.iisp.bind.enums.BindType;
import com.sf.iisp.bind.enums.State;
import com.sf.iisp.kafka.adapter.IKafkaProducerPoolDelegate;
import com.sf.iisp.util.DateUtils;
import com.sf.kafka.check.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.Resource;
import java.util.Date;
import java.util.UUID;

/**
 * 预约解绑号码的 Redis 消息监听器
 * Created by qi on 16-7-6.
 */
public class UnbindingListener implements MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(UnbindingListener.class);

    private RedisTemplate<String, Object> redisTemplate;
    private ListOperations<String, Object> listOps;

    @Resource
    private BindingBiz bindingBiz;

    @Resource
    private IBindingDao bindingDao;

    @Resource
    private UnbindingBiz unbindingBiz;

    @Resource
    private IUnbindingDao unbindingDao;

    @Qualifier("producer")
    @Resource
    private IKafkaProducerPoolDelegate delegate;

    @Value("isSendKafka:false")
    private String isSend;

    public UnbindingListener(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.listOps = redisTemplate.opsForList();
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        byte[] body = message.getBody();
        byte[] channel = message.getChannel();

        String topic = redisTemplate.getStringSerializer().deserialize(channel);
        String itemValue = (String) redisTemplate.getValueSerializer().deserialize(body);
        long size = listOps.size(itemValue);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("topic = {}, unbinding's key  = {}", topic, itemValue);
            LOGGER.debug("receive message size = {}", size);
        }

        for (long i = 0; i < size; i++) {
            BindingRecord br = (BindingRecord) listOps.rightPop(itemValue);
            BindingRecord historyBr = bindingDao.findBindRerordByTransId(br.getTransId());
            if (br != null && historyBr != null) {
                ReturnBindingInfo returnInfo = bindingBiz.convertBindingRecord2ReturnInfo(historyBr);

                String courierPhone = historyBr.getCourierPhone();
                String customerPhone = historyBr.getCustomerPhone();
                String virtualNo = historyBr.getVirtualNo();
                String transId = historyBr.getTransId();

                UnbindingRecord ur = unbindingDao.findByStateAndPhones(State.RESERVED.toString(), courierPhone, customerPhone);
                if (null == ur) {
                    ur = new UnbindingRecord();
                    ur.setMsgType(BindType.UNBIND.toString());
                    ur.setSysCode(historyBr.getSysCode());
                    ur.setTransId(UUID.randomUUID().toString());
                    ur.setTransTimestamp(DateUtils.date2Str(new Date(), DateUtils.datetimeFormat));
                    ur.setCourierPhone(courierPhone);
                    ur.setCustomerPhone(customerPhone);
                    ur.setExtendField1(historyBr.getExtendField1());
                    ur.setExtendField2(historyBr.getExtendField2());
                    ur.setOriginTransId(historyBr.getTransId());
                }


                try {
                    Boolean isUnbind = unbindingBiz.unbind(courierPhone, customerPhone, virtualNo, transId, ur);
                    if (isUnbind) {
                        LOGGER.info("reserve unbinding record :  courierPhone = {}, customerPhone = {} , virtual no ={}", historyBr.getCourierPhone(), historyBr.getCustomerPhone(), virtualNo);
                        ur.setMsg(State.SUCCESS.getMsg());
                        ur.setState(State.SUCCESS.toString());
                        returnInfo.setState(State.SUCCESS.toString());
                        returnInfo.setMsg(State.SUCCESS.getMsg());
                        historyBr.setState(State.UNBIND.toString());
                    }
                } catch (BindingException e) {
                    LOGGER.error(e.getMessage(), e);
                    returnInfo.setState(e.getErrorCode());
                    returnInfo.setMsg(e.getMessage());

                    //TODO:解绑失败之后，是否更新现有绑定状态？
                    historyBr.setState(e.getErrorCode());
                    ur.setMsg(e.getMessage());
                    ur.setState(e.getErrorCode());
                } finally {
                    String msg = JsonUtil.writeValueAsString(returnInfo);
                    LOGGER.info("send reserve unbind message = {}", msg);

                    //TODO:测试关闭！！
                    if ("true".equals(isSend)) {
                        delegate.sendString(msg);
                    }

                    ur.setVirtualNo(virtualNo);

                    ur.setUnbindTime(new Date());

                    bindingDao.updateSaveBindRecord(historyBr);
                    unbindingDao.saveOrUpdateUnbindRecord(ur);
                }
            }

        }
    }
}
